package com.shujia.card

import java.util

import com.alibaba.fastjson.JSON
import com.google.gson.Gson
import com.shujia.util.{CarUtil, SparkStreamTool}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import redis.clients.jedis.Jedis

object HaveBeenSupervisedChecking extends SparkStreamTool {
  override def run(args: Array[String]): Unit = {

    //读取卡口过车数据
    val carsDS: DStream[CarUtil.Car] = CarUtil.loadKafkaCar(ssc, "HaveBeenSupervisedChecking3")

    //获取布控列表
    /**
      * 在redis中保存需要布控的车辆列表
      *
      * sadd check_list 皖A9A7N2
      * sadd check_list 皖AR3HY0
      * sadd check_list 皖ANQEQ9
      *
      */

    //此处的代码只会执行一次

    val checkCar: DStream[CarUtil.Car] = carsDS.transform(rdd => {

      //此处的代码每一个batch都会执行一次

      //将java集合转换成scala集合的隐式转换
      import scala.collection.JavaConversions._
      //将scala集合转换成java集合的隐式转换
      //import scala.collection.JavaConverters._

      val jedis = new Jedis("master", 6379)
      //布控列表
      val checkList: Set[String] = jedis.smembers("check_list").toSet

      println("布控列表：" + checkList)

      //将布控列表广播出去
      val broCheckList: Broadcast[Set[String]] = sc.broadcast(checkList)

      //过滤数据
      val filterRDD: RDD[CarUtil.Car] = rdd.filter(car => {
        val carId: String = car.car

        //判断车辆是否是布控列表中的车辆
        broCheckList.value.contains(carId)
      })

      //清空广播变量
      broCheckList.unpersist()

      //返回监控到的数据
      filterRDD
    })



    //将布控到的车辆保存到redis中
    checkCar.foreachRDD(rdd => {
      rdd.foreachPartition(iter => {

        //创建redis链接
        val jedis = new Jedis("master", 6379)

        //换一个库保存结果
        jedis.select(1)

        iter.foreach(car => {
          /**
            * 以车牌号作为key ,使用list作为value,  value 保存车辆过车的一条数据，以json格式保存
            *
            */

          val gson = new Gson()
          val carJson: String = gson.toJson(car)

          val key: String = "HaveBeenSupervisedChecking:" + car.car

          jedis.lpush(key, carJson)

        })

        jedis.close()
      })
    })

  }
}
